<?php
/**
 * Copyright (c) [2019] [吴跃忠]
 * [selibra] is licensed under the Mulan PSL v1.
 * You can use this software according to the terms and conditions of the Mulan PSL v1.
 * You may obtain a copy of Mulan PSL v1 at:
 * http://license.coscl.org.cn/MulanPSL
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
 * PURPOSE.
 * See the Mulan PSL v1 for more details.
 */

namespace LeoMQ\Realizes\RabbitMQ;

use Co\Channel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use Selibra\Config\Config;
use Selibra\Tools\Console;
use Swoole\Coroutine;

class ConnectionPool
{
    /**
     * 连接池
     * @var Channel
     */
    protected Channel $pool;

    /**
     * @var static
     */
    protected static self $ins;


    /**
     * @return static
     */
    public static function getIns()
    {
        if (!isset(self::$ins)) {
            self::$ins = new self();
            self::$ins->genConnects();
        }
        return self::$ins;
    }


    /**
     * ConnectionPool constructor.
     */
    private function __construct()
    {

    }


    /**
     * 生成连接
     */
    private function genConnects()
    {
        // 生成连接池
        if (!isset($this->pool)) {

            $poolSize = Config::get('LeoMQ.pool_size') ?: 20;
            $this->pool = new Channel($poolSize);
            $connect = new Connection();
            for ($i = 0; $i < $poolSize; $i++) {
                $amqpConnection = $connect->getConnection();
                $this->pool->push($amqpConnection);
            }
        }
    }


    /**
     * @return mixed
     */
    public function get()
    {

        $timeout = Config::get('application.db.pool_get_timeout');
        /** @var AMQPStreamConnection $amqpConnection */
        $amqpConnection = $this->pool->pop($timeout);
        // 检测连接是否可用
        $amqpConnection = $this->verifyAvailabilityRestart($amqpConnection);

        // 协程结束，释放连接
        defer(function () use ($amqpConnection) { // 释放
            $this->verifyAvailabilityRestart($amqpConnection);
            $this->pool->push($amqpConnection);
        });
        return $amqpConnection;
    }


    /**
     * @param $amqpConnection
     */
    public function verifyAvailabilityRestart($amqpConnection)
    {
        if (!$amqpConnection->isConnected() || $amqpConnection->isBlocked()) {
            $connect = new Connection();
            $amqpConnection = $connect->getConnection();
        }
        return $amqpConnection;
    }
}